package com.blockmeta.bbs.businesslibrary.net.rxjava;

import com.blockmeta.bbs.businesslibrary.net.pojo.BaseCodeResponse;
import com.blockmeta.bbs.businesslibrary.net.pojo.BaseImResponse;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

/**
 * Created by author
 * Date 2019/8/7
 */
public class ObservableRxUtil {

    /**
     * 简化线程操作
     */
    public static <T> ObservableTransformer<T, T> rxSchedulerHelper() {
        return new ObservableTransformer<T, T>() {
            @Override
            public ObservableSource<T> apply(Observable<T> upstream) {
                return upstream.subscribeOn(Schedulers.io())
                               .observeOn(AndroidSchedulers.mainThread());
            }
        };
    }

    /**
     * 统一返回结果处理
     *
     * @param <T>
     * @return
     */
    public static <T> ObservableTransformer<BaseImResponse<T>, T> commonResult() {   //compose判断结果
        return new ObservableTransformer<BaseImResponse<T>, T>() {
            @Override
            public Observable<T> apply(Observable<BaseImResponse<T>> commonResponseFlowable) {
                return commonResponseFlowable.flatMap(new Function<BaseImResponse<T>, Observable<T>>() {
                    @Override
                    public Observable<T> apply(BaseImResponse<T> tBaseCodeResponse) {
                        if ("200".equals(tBaseCodeResponse.code) && tBaseCodeResponse.data != null) {
                            return createObservable(tBaseCodeResponse.data);
                        } else {
                            return Observable.error(new CustomException(tBaseCodeResponse.code, tBaseCodeResponse.msg));
                        }
                    }
                });
            }
        };
    }


    public static <T> ObservableTransformer<BaseCodeResponse<T>, T> commonResultByCode() {
        return new ObservableTransformer<BaseCodeResponse<T>, T>() {
            @Override
            public ObservableSource<T> apply(Observable<BaseCodeResponse<T>> upstream) {
                return upstream.flatMap(new Function<BaseCodeResponse<T>, Observable<T>>() {
                    @Override
                    public Observable<T> apply(BaseCodeResponse<T> tBaseCodeResponse) throws Exception {
                        if ("200".equals(tBaseCodeResponse.code) && tBaseCodeResponse.data != null) {
                            return createObservable(tBaseCodeResponse.data);
                        } else {
                            return Observable.error(new CustomException(tBaseCodeResponse.code, tBaseCodeResponse.msg));
                        }
                    }
                });
            }
        };
    }

    /**
     * 创建Observable
     *
     * @param data
     * @param <T>
     * @return
     */
    private static <T> Observable<T> createObservable(final T data) {
        return Observable.create(new ObservableOnSubscribe<T>() {
            @Override
            public void subscribe(ObservableEmitter<T> emitter) throws Exception {
                try {
                    emitter.onNext(data);
                    emitter.onComplete();
                } catch (Exception e) {
                    emitter.onError(e);
                }
            }
        });

    }


}
